自定义触发器
相关网址
摩天轮网址
触发器代码（指定数据量或时间触发）
public class CountAndTimeTrigger<W extends Window> extends Trigger<Object, W> {
private final long size;
private final long interval;
private final ReducingStateDescriptor<Integer> countStateDesc = new ReducingStateDescriptor<>("count", new ReduceSum(), IntSerializer.INSTANCE);
private final ReducingStateDescriptor<Long> timeStateDesc = new ReducingStateDescriptor<>("fire-interval", new ReduceMin(), LongSerializer.INSTANCE);
public CountAndTimeTrigger(long size, long interval) {
this.size = size;
this.interval = interval;
}
@Override
public TriggerResult onElement(Object o, long timestamp, W window, TriggerContext ctx) throws Exception {
ReducingState<Integer> count = ctx.getPartitionedState(countStateDesc);
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
count.add(1);
if (count.get() >= size) {
log.info("窗口结束，计数器触发：" + count.get());
count.clear();
if (null != fireTimestamp.get() && fireTimestamp.get() != window.maxTimestamp()) {
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
}
fireTimestamp.clear();
return TriggerResult.FIRE_AND_PURGE;
}
timestamp = ctx.getCurrentProcessingTime();
if (fireTimestamp.get() == null) {
long nextFireTimeStamp = timestamp + interval;
ctx.registerProcessingTimeTimer(nextFireTimeStamp);
fireTimestamp.add(nextFireTimeStamp);
}
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
ReducingState<Integer> count = ctx.getPartitionedState(countStateDesc);
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
if (fireTimestamp.get() != null && time == window.maxTimestamp()) {
log.info("窗口正常结束：" + time);
// 窗口结束，清0条数和时间的计数器
count.clear();
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
fireTimestamp.clear();
return TriggerResult.FIRE_AND_PURGE;
} else if (fireTimestamp.get() != null && fireTimestamp.get().equals(time)) {
log.info("窗口结束，定时器触发：" + time);
// 时间计数器触发，清0条数和时间计数器
count.clear();
fireTimestamp.clear();
return TriggerResult.FIRE_AND_PURGE;
}
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long l, W w, TriggerContext triggerContext) throws Exception {
return TriggerResult.CONTINUE;
}
@Override
public void clear(W w, TriggerContext ctx) throws Exception {
ctx.getPartitionedState(countStateDesc).clear();
ctx.getPartitionedState(timeStateDesc).clear();
}
@Override
public void onMerge(W window, OnMergeContext ctx) throws Exception {
super.onMerge(window, ctx);
ctx.mergePartitionedState(timeStateDesc);
ctx.mergePartitionedState(countStateDesc);
}
}
class ReduceSum implements ReduceFunction<Integer> {
@Override
public Integer reduce(Integer t0, Integer t1) throws Exception {
return t0 + t1;
}
}
class ReduceMin implements ReduceFunction<Long> {
@Override
public Long reduce(Long t0, Long t1) throws Exception {
return Math.min(t0, t1);
}
}
本地比较稳定配置
batchSize=3500
batchDelay=10000
测试代码
public class TriggerTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<EventLog> ds = env.addSource(new MySourceFunction());
// Random random = new Random();
// ds.map(event -> new Tuple2<Integer, EventLog>(random.nextInt(10), event)).returns(new TypeHint<Tuple2<Integer, EventLog>>() {
// })
// .keyBy(i->i.f0)
// .window(GlobalWindows.create())
// .trigger(new MaxRecordCountAndProcessingTimeTrigger<>(500, 10000))
// .process(new ProcessWindowFunction<Tuple2<Integer, EventLog>, String, Integer, GlobalWindow>() {
// @Override
// public void process(Integer integer, ProcessWindowFunction<Tuple2<Integer, EventLog>, String, Integer, GlobalWindow>.Context context, Iterable<Tuple2<Integer, EventLog>> iterable, Collector<String> collector) throws Exception {
// int size = 0;
// for (Tuple2<Integer, EventLog> tuple : iterable) {
// size++;
// }
// String str = "current key is: " + integer + " current size is: " + size;
// collector.collect(str);
// }
// }).print();
ds.keyBy(EventLog::getEventId)
.window(GlobalWindows.create())
.trigger(new CountAndTimeTrigger<>(500, 5000))
// .trigger(new MaxRecordCountAndProcessingTimeTrigger<>(500, 10000))
.process(new ProcessWindowFunction<EventLog, String, String, GlobalWindow>() {
@Override
public void process(String s, ProcessWindowFunction<EventLog, String, String, GlobalWindow>.Context context, Iterable<EventLog> iterable, Collector<String> collector) throws Exception {
int size = CollectionsUtil.iterableToList(iterable).size();
collector.collect("current key is: " + s + " current size is: " + size);
}
}).setParallelism(1).print();
env.execute();
}
}
class MySourceFunction implements SourceFunction<EventLog> {
volatile boolean flag = true;
volatile int count = 0;
@Override
public void run(SourceContext<EventLog> ctx) throws Exception {
EventLog eventLog = new EventLog();
// String[] events = {"appLaunch", "pageLoad", "adShow", "adClick", "itemShare", "itemCollect", "putBack", "wakeUp", "appClose"};
String[] events = {"appLaunch", "pageLoad", "adShow"};
HashMap<String, String> eventInfoMap = new HashMap<>();
for (int i = 0; i < 600; i++) {
eventLog.setGuid(RandomUtils.nextLong(1, 1000));
eventLog.setSessionId(RandomStringUtils.randomAlphabetic(12).toUpperCase());
eventLog.setTimeStamp(System.currentTimeMillis());
eventLog.setEventId(events[2]);
eventInfoMap.put(RandomStringUtils.randomAlphabetic(1), RandomStringUtils.randomAlphabetic(2));
eventLog.setEventInfo(eventInfoMap);
ctx.collect(eventLog);
}
eventLog.setGuid(RandomUtils.nextLong(1, 1000));
eventLog.setSessionId(RandomStringUtils.randomAlphabetic(12).toUpperCase());
eventLog.setTimeStamp(System.currentTimeMillis());
eventLog.setEventId(events[0]);
eventInfoMap.put(RandomStringUtils.randomAlphabetic(1), RandomStringUtils.randomAlphabetic(2));
eventLog.setEventInfo(eventInfoMap);
ctx.collect(eventLog);
while (flag) {
eventLog.setGuid(RandomUtils.nextLong(1, 1000));
eventLog.setSessionId(RandomStringUtils.randomAlphabetic(12).toUpperCase());
eventLog.setTimeStamp(System.currentTimeMillis());
eventLog.setEventId(events[RandomUtils.nextInt(0, events.length)]);
eventInfoMap.put(RandomStringUtils.randomAlphabetic(1), RandomStringUtils.randomAlphabetic(2));
eventLog.setEventInfo(eventInfoMap);
ctx.collect(eventLog);
eventInfoMap.clear();
count++;
if (count % 100 == 0) {
Thread.sleep(20000);
break;
}
}
}
@Override
public void cancel() {
flag = false;
}
}
相关案例
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @Auther WeiJiQian
* @描述 CountAndTimeTrigger : 满足一定条数和时间触发
* * 条数的触发使用计数器计数
* * 时间的触发，使用 flink 的 timerServer，注册触发器触发
*/
public class CountAndTimeTrigger<W extends Window> extends Trigger<Object, W> {
private Logger logger = LoggerFactory.getLogger(this.getClass());
// 触发的条数
private final long size;
// 触发的时长
private final long interval;
private static final long serialVersionUID = 1L;
// 条数计数器
private final ReducingStateDescriptor<Integer> countStateDesc =
new ReducingStateDescriptor<>("count", new ReduceSum(), IntSerializer.INSTANCE);
// 时间计数器，保存下一次触发的时间
private final ReducingStateDescriptor<Long> timeStateDesc =
new ReducingStateDescriptor<>("fire-interval", new ReduceMin(), LongSerializer.INSTANCE);
public CountAndTimeTrigger(long size, long interval) {
this.size = size;
this.interval = interval;
}
// 每条元素到来时.
@Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
// 注册窗口结束的触发器, 不需要会自动触发
// ctx.registerProcessingTimeTimer(window.maxTimestamp());
// count
ReducingState<Integer> count = ctx.getPartitionedState(countStateDesc);
//interval
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
// 每条数据 counter + 1
count.add(1);
if (count.get() >= size) {
System.out.println("窗口结束: 计数器触发 count : {}"+ count.get());
// 满足条数的触发条件，先清 0 条数计数器
count.clear();
// 满足条数时也需要清除时间的触发器，如果不是创建结束的触发器
if (fireTimestamp.get() != window.maxTimestamp()) {
// logger.info("delete trigger : {}, {}", sdf.format(fireTimestamp.get()), fireTimestamp.get());
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
}
fireTimestamp.clear();
// fire 触发计算
return TriggerResult.FIRE;
}
// 触发之后，下一条数据进来才设置时间计数器注册下一次触发的时间
timestamp = ctx.getCurrentProcessingTime();
// timestamp = System.currentTimeMillis();
if (fireTimestamp.get() == null) {
// long start = timestamp - (timestamp % interval);
long nextFireTimestamp = timestamp + interval;
// logger.info("register trigger : {}, {}", sdf.format(nextFireTimestamp), nextFireTimestamp);
ctx.registerProcessingTimeTimer(nextFireTimestamp);
fireTimestamp.add(nextFireTimestamp);
}
return TriggerResult.CONTINUE;
}
// 处理时间到的时候,开始处理
@Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
// count
ReducingState<Integer> count = ctx.getPartitionedState(countStateDesc);
//interval
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
// time trigger and window end
if (fireTimestamp.get() != null && time == window.maxTimestamp()) {
System.out.println("窗口结束: 正常结束 {}" + time);
// 窗口结束，清0条数和时间的计数器
count.clear();
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
fireTimestamp.clear();
return TriggerResult.FIRE_AND_PURGE;
} else if (fireTimestamp.get() != null && fireTimestamp.get().equals(time)) {
System.out.println("窗口结束:时间计数器触发, time : {}" + time);
// 时间计数器触发，清0条数和时间计数器
count.clear();
fireTimestamp.clear();
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception {
// count
ReducingState<Integer> count = ctx.getPartitionedState(countStateDesc);
//interval
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
// time trigger and window end
if (time == window.maxTimestamp()) {
System.out.println("窗口结束 : {}"+ time);
// 窗口结束，清0条数和时间的计数器
count.clear();
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
fireTimestamp.clear();
return TriggerResult.FIRE_AND_PURGE;
} else if (fireTimestamp.get() != null && fireTimestamp.get().equals(time)) {
System.out.println("时间计数器触发, time : {}"+ time);
// 时间计数器触发，清0条数和时间计数器
count.clear();
fireTimestamp.clear();
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
@Override
public void clear(W window, TriggerContext ctx) throws Exception {
ctx.getPartitionedState(countStateDesc).clear();
ctx.getPartitionedState(timeStateDesc).clear();
}
// 多个slot 中的 数据合并.
@Override
public void onMerge(W window, OnMergeContext ctx) throws Exception {
super.onMerge(window, ctx);
ctx.mergePartitionedState(timeStateDesc);
ctx.mergePartitionedState(countStateDesc);
}
}